#include "FirstPhase.hpp"

int FirstPhase::run() {
    // Read the data file. Parse it to have the transactions vector and also
    // the items-counts hash.
    clock_t start = clock();
    if (findLocalCounts() != 0) {
        cerr << "Error reading file. (Is data file really there?)" << endl;
        return -1;
    }

    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": Input file parsed -\t" 
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }

    // Local counts have been found. Reduce counts and prune those with
    // lower than minimum support.
    start = clock();
#ifdef USE_MPI
    if (Globals::reduceBroadcast)
        reduceAndBcastCounts();
    else
        allReduceCounts();
#else
    reducedCounts = itemCounts;
    itemCounts.clear();
    pruneCounts();
#endif

    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": Local counts reduced & pruned -\t" 
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }

    // Now that we have global counts, prune the transactions if the option
    // is specified to prune transactions before sorting them.
    if (Globals::pruneMethod != 1) {
        start = clock();
        pruneTransactions();
        cout << "Transactions pruned before sorting:\t"
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }

    // Sort local transactions.
    start = clock();
    sortLocal();
    
    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": Transactions sorted -\t"
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }

    // transactions are now sorted. (it was a vector of pointers to string 
    // vectors). We now prune items with support less than the minimum (if 
    // option is specified to prune transactions after sorting them).
    if (Globals::pruneMethod == 1) {
        start = clock();
        pruneTransactions();
        if (Globals::verbose) {
            cout << "Rank " << Globals::rank << ": Transactions pruned -\t"
                << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
        }
    }

    start = clock();
    // construct fp-trees by reading each transaction according to this list.
    buildFpTree();
    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": FP Tree built -\t"
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }

    // Each node has local fp-trees. Distribute and combine them. ie. reduce.
    start = clock();
    // Say we have two processes in total. The first one (with rank 0) is the 
    // master, he will receive second process' fp-tree, reduce and send it back.
#ifdef USE_MPI
  #ifndef SIMPLE_REDUCE
    if (Globals::reduceBroadcast)
        reduceAndBcastTrees();
    else
        all_reduce(world, fpTree, mergedTree, reduce_trees());
  #else
    int tag = 9;
    if (Globals::rank == 1) {
        world.send(0, tag, fpTree);
        world.recv(0, tag, mergedTree);
    }
    else {
        world.recv(1, tag, mergedTree);
        mergedTree.merge(fpTree);
        world.send(1, tag, mergedTree);
    }
  #endif
#else
    // Serial version
    mergedTree = fpTree;
#endif
    // We don't need fpTree anymore, Clear it.
    fpTree.makeEmpty();

    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": FP Tree reduced -\t"
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }
    // Store reduced counts in the merged tree.
    mergedTree.setItemCounts(reducedCounts);
}

int FirstPhase::findLocalCounts() {
	string line;
	ifstream datafile(Globals::inputFileName.c_str());

	if (!datafile.is_open()) {
        cout << "Input file couldn't be opened! Are you sure it is there?" 
            << endl;
        return -1;
    }

    // if (Globals::rank == 0)
    //    Globals::debug();
    if (Globals::dataType == 1) {
        // First of all, go to start line.
        for (int lineNumber = 1; lineNumber < startXact; lineNumber++)
            if (!getline (datafile, line)) {
                cout << "Unexpected end of file." << endl;
                return -1;
            }

        typedef tokenizer<> MyTok;

        for ( int i = partSize; i > 0; i--) {
            if (!getline (datafile, line)) {
                cout << "Unexpected end of file." << endl;
                return -1;
            }
            vector<string> * pStringVector = new vector<string>;

            MyTok tok(line);
            for(MyTok::iterator beg=tok.begin(); beg != tok.end(); beg++) {
                // if no previous record exists, create it.
                itemCounts[*beg]++;
                // Add this item to the transaction vector.
                pStringVector->push_back(*beg);
            }

            // Add this string vector transaction to the local DB.
            transactions->push_back(pStringVector);
        }
    }
    else { // Parse IBM type of data file, generated using Quest.
        int xact = 1;
        int nextXact;
        string item;
        // First of all, go to start line.
        stringstream ss;
        for ( ; ; ) {
            if (!getline(datafile, line)) {
                cout << "Unexpected end of file." << endl;
                return -1;
            }
            ss.clear();
            ss.str(line);
            ss >> xact;
            if (xact >= startXact)
                break;
        }
        // I have read the cust_id part of the beginning xact.
        for (int i = partSize; i > 0; i--) {
            vector<string> * pStringVector = new vector<string>;
            for ( ; ; ) {
                // Read the xact_id part.
                ss >> xact;
                // Read the item
                ss >> item;
                itemCounts[item]++;
                pStringVector->push_back(item);

                // Go for the next line. ie. Next transaction.
                if (!getline(datafile, line)) {
                    if (i == 1)
                        break;
                    else {
                        cout << "Unexpected end of file." << endl;
                        return -1;
                    }                        
                }
                ss.clear();
                ss.str(line);
                // Get cust_id part of next xact.
                ss >> nextXact;
                if (nextXact != xact)
                    break;
            }

            // Add this string vector transaction to the local DB.
            transactions->push_back(pStringVector);
        }
    }

    // Close the file in the end.
    datafile.close();
    return 0;
}

#ifdef USE_MPI
void FirstPhase::reduceAndBcastCounts() {
    clock_t start = clock();
    // Let root be 0?
    int master = 0;
    reduce(world, itemCounts, reducedCounts, reduce_counts(), master);
    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": Reduce and broadcast icinde reduce metodu - "
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }
    // We don't need itemCounts no more. Clear it.
    itemCounts.clear();
    if (Globals::rank == master) {
        // prune only on the master since only that node
        // has the global counts.
        pruneCounts();
    }
    // Send reduced counts to everyone else.
    start = clock();
    broadcast(world, reducedCounts, master);
    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": Reduce and broadcast icinde broadcast metodu - "
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }
}

void FirstPhase::reduceAndBcastTrees() {
    // Let root be 0.
    clock_t start = clock();
    int master = 0;
    reduce(world, fpTree, mergedTree, reduce_trees(), master);
    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": Reduce and broadcast trees icinde reduce metodu - "
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }
    // Send reduced tree to everyone else.
    start = clock();
    broadcast(world, mergedTree, master);
    if (Globals::verbose) {
        cout << "Rank " << Globals::rank << ": Reduce and broadcast trees icinde broadcast metodu - "
            << (clock() - start) / double (CLOCKS_PER_SEC) << " s." << endl;
    }
}

void FirstPhase::allReduceCounts() {
    all_reduce(world, itemCounts, reducedCounts, reduce_counts());
    // We don't need itemCounts no more. Clear it.
    itemCounts.clear();
    // Prune according to minimum support on every node.
    pruneCounts();
}
#endif

void FirstPhase::pruneCounts() {
    hash_map<string, int>::iterator it;
    for (it = reducedCounts.begin(); it != reducedCounts.end(); ) {
        if (it->second < Globals::minimumSupport)
            reducedCounts.erase(it++);
        else 
            ++it;
    }
}

void FirstPhase::pruneTransactions() {
    // Prune transactions based on supports of its items. Only the transactions
    // local to the node are pruned. It may be pruned either before sorting them
    // or after that.
    vector< vector<string> *>::iterator it = transactions->begin(), 
        endIt = transactions->end();
    if (Globals::pruneMethod == 1) {
        // FIXME: Transactiondaki itemlarin tamami min supporttan kucuk oldugunda silinmemisler.
        for ( ; it != endIt; it++) {
            vector<string>::iterator itemsEnd = (*it)->end(), itemsBegin = (*it)->begin();
            itemsEnd--;
            while (itemsEnd >= itemsBegin 
                    && reducedCounts[*itemsEnd] < Globals::minimumSupport) {
                itemsEnd--;
            }
            itemsEnd++;

            // Erase all elements with less than minimum support.
            // No need to remove an empty string vector pointer as it will have
            // an overhead of moving later pointers. 
            (*it)->erase(itemsEnd, (*it)->end());
        }
    }
    else {
        vector< vector<string> * > * sortedTransactions = new vector< vector<string> * >;
        for ( ; it != endIt; it++) {
            vector<string>::iterator trIt = (*it)->begin(), 
                trEnd = (*it)->end();
            vector<string> * sortedTransaction = new vector<string>;
            for ( ; trIt != trEnd; trIt++) {
                if (reducedCounts[*trIt] >= Globals::minimumSupport)
                    sortedTransaction->push_back(*trIt);
            }

            // Check if all the items are pruned, i.e. they all have support less than 
            // minimum support. Do not insert an empty string vector pointer.
            if (sortedTransaction->empty())
                delete sortedTransaction;
            else
                sortedTransactions->push_back(sortedTransaction);
        }

        // Sorted transactions now has it sorted. Change this with the 
        // unsorted one.
        delete transactions;
        transactions = sortedTransactions;
    }
}

void FirstPhase::printCounts() {
    hash_map<string, int>::iterator it;
    cout << "Counts for the items to follow." << endl;

    for ( it = reducedCounts.begin(); it != reducedCounts.end(); it++) {
        if (it->second > 0) {
        cout << it->first << ": " << it->second << " number of times."
            << endl;
        }
    }

    cout << "Counts for the items finished." << endl;
}

void FirstPhase::sortLocal() {
    vector< vector<string> * >::iterator localIt = transactions->begin();

    for( ; localIt != transactions->end(); localIt++) {
        sort((*localIt)->begin(), (*localIt)->end(), SortingFunctor(&(reducedCounts)));
    }
}

void FirstPhase::printTransactions() {
    vector< vector<string> * >::iterator localIt = transactions->begin();
    cout << "Transactions for this local node to follow. Rank: " << Globals::rank << endl;

    for( ; localIt != transactions->end(); localIt++) {
        vector<string>::iterator trIt = (*localIt)->begin();
        for ( ; trIt != (*localIt)->end(); trIt++)
            cout << *trIt << " ";
        cout << endl;
    }

    cout << "Transactions for this local node finished." << endl;
}

// Builds fp-tree by inserting each transaction to the tree one by one.
void FirstPhase::buildFpTree() {
    vector< vector<string> * >::iterator localIt = transactions->begin(),
        end = transactions->end();

    for( ; localIt != end; localIt++) {
        fpTree.insert(*localIt);
    }
}

// Gives back any memory we don't need in second phase. The ones not deleted are
// the merged fp-tree and the reduced item counts.
void FirstPhase::clear() {
    // Remove all the elements from transactions. We won't need each transaction
    // since it will be compressed in an fp-tree anyway.
    vector< vector<string> * >::iterator it = transactions->begin(),
        end = transactions->end();

    for ( ; it != end; it++) {
        delete *it;
    }

    // Clear itemCounts as well, since we have reducedCounts.
    itemCounts.clear();
    // Clear local fp-tree:
    fpTree.makeEmpty();
}

